/*
 * Copyright (C) 2014 Qloudgen Technology
*/



import org.apache.log4j.PropertyConfigurator;

import com.qloudgen.csb.client.Connection;
import com.qloudgen.csb.client.Consumer;
import com.qloudgen.csb.client.KfkProducer;
import com.qloudgen.csb.client.Message;
import com.qloudgen.csb.client.MessageListener;
import com.qloudgen.csb.client.Queue;
import com.qloudgen.csb.client.Session;

//public class Provider implements KfkMessageListener {  
public class Server implements MessageListener {  
	
	long count = 0;
	static KfkProducer sender;
	
	private Server() {
	}
	
	@Override
	//public void onMessage(String message, long offset) {
	public void onMessage(Message message) {
		//System.out.println(message);
		//try {
		//	Thread.sleep(2);
		//} catch (InterruptedException e) {
		//	e.printStackTrace();
		//}
		
		count++;
		if (count >= 1000) {
			System.out.println(count + " messages processed...");
			count = 0;
		}
		
		sender.send("reply", "done");
	}

    public static void main(String[] args) throws Exception {
    	PropertyConfigurator.configure("conf/log4j.properties");
    	System.out.println("providing...");
    	
		sender = new KfkProducer(args[1], -1, null);

    	try {
    		Connection connection = new Connection();
    		Session session = connection.createSession();
    		Queue queue = session.createQueue("qbank", args[0]);
    		Consumer consumer = session.createConsumer(queue);
    		consumer.setMessageListener(new Server());
    		
    		connection.start();
    		//KfkConsumer kc = new KfkConsumer(args[0], args[1],
    		//		args[2], 10, null);
    		//kc.setMessageListener(new Provider());
    		//kc.run();
    		//new Thread(new Provider("tcp://" + args[0], null)).start();
    	} catch (Exception e) {
    		e.printStackTrace();
    	}
    	
    	System.out.println("running...");
    	//System.exit(0);
    }
}

